package sc.analysis.movingavg;



import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import sc.writable.PriceData;
import sc.writable.WritableArrayListFloat;



/**
 * Mapper of the Moving Average analysis. Calculate the moving average from
 * price data.
 * 
 * @author Quentin Rossé
 *
 */
public class MaMapper extends Mapper<Text, PriceData, Text, MaData>{
	
	private static final int MA_10  = 10;
	private static final int MA_20  = 20;
	private static final int MA_50  = 50;
	private static final int MA_200 = 200;
	
	public void map(Text key, PriceData value, Context context) throws IOException, InterruptedException{

		WritableArrayListFloat closePrices = value.getClosePrices();
		int arraySize = closePrices.size();
		float sum = 0.f;
		int num = 0;
		
		for(int i=arraySize-1; i>=0; i--){
			sum += closePrices.getFloat(i);
			num++;
			if((num == MA_10 || num == MA_20 || num == MA_50 || num == MA_200) &&
					arraySize > num){
				context.write(key, new MaData(sum/num, //current moving average
						                      (sum - closePrices.getFloat(arraySize-1) + closePrices.getFloat(i-1)) / num, //previous average
						                      closePrices.getFloat(arraySize-1), //last price (last close)
						                      closePrices.getFloat(arraySize-2), //previous price (close the day before)
						                      num)); //number of days for the moving average
				System.out.println("SMA" + num +"  " + sum/num + "   oldSMA : " +  (sum - closePrices.getFloat(arraySize-1) + closePrices.getFloat(i-1)) / num);
			}
		}
		
		
	}
	
	@Override
	public void setup(Context context){}
	
	@Override
	public void cleanup(Context context){}
	
}
